/*
 * ObjectMessageProtocolHandler.java    0.0.1    26/set/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */
package org.mentalsmash.tazio.net.cs.server;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;

import org.mentalsmash.tazio.net.cs.MessageInputStream;
import org.mentalsmash.tazio.net.cs.MessageOutputStream;
import org.mentalsmash.tazio.net.cs.RuntimeCSSubsystemException;
import org.mentalsmash.tazio.net.cs.protocol.MessageHeader;
import org.mentalsmash.tazio.net.cs.protocol.NIOObjectContainer;
import org.mentalsmash.tazio.net.cs.server.impl.ChannelFacade;
import org.mentalsmash.tazio.net.cs.server.impl.InputQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The <tt>ObjectMessageProtocolHandler</tt> class provides an
 * <code>abstract</code> implementation of {@link ProtocolHandler}, providing
 * the basic abstractions to receive and send objects that implement the
 * {@link Serializable} interface over a channel, instead of being force to cope
 * with <code>byte[]</code> or <code>Buffer</code>s.
 * <p>
 * Concrete classes that want to subclass <tt>ObjectMessageProtocolHandler</tt>
 * have to implement the
 * {@link #handleMessage(Serializable,MessageOutputStream)} method ( along with
 * the {@link #registered(ChannelFacade)} and
 * {@link #unregistered(ChannelFacade)} from the {@link ProtocolHandler}
 * interface).
 * </p>
 * <p>
 * {@link #handleMessage(Serializable,MessageOutputStream)} will be
 * automatically called each time an object is received from the channel. That
 * object is passed as argument of the method, together with a reference to an
 * {@link MessageOutputStream} instance, that provides convenience methods to
 * write back the responce to the channel that originated the request.
 * </p>
 * 
 * @version 0.0.1 26/set/2009
 * @author Andrea Reale <andrea.reale@gmail.com>
 * @see ProtocolHandler
 * @see ObjectChannelWriter
 */
public abstract class ObjectMessageProtocolHandler implements DeflatingProtocolHandler {

    private final static Logger log = LoggerFactory
	    .getLogger(ObjectMessageProtocolHandler.class);

    private MessageHeader nextHeader;
    private ByteBuffer headerBuffer;

    // the next linked list accumulates output from the input queue waiting for
    // a
    // complete payload to be seen
    // private LinkedList<ByteBuffer> queue;
    private ByteBuffer partialInput = null;
    // private int payloadCurrentSize = 0;

    private boolean headerReceived = false;

    //true for deflating being active
    private boolean deflating = true;

    // /**
    // * Enqueues at max howMuch bytes into the queue of ByteBuffers. If input
    // * containes less then howMuch bytes (say r bytes), enques only r bytes
    // *
    // * @param input
    // * @param howMuch
    // * @return the number of bytes enqueued
    // *
    // */
    // private int enqueue(final InputQueue input, final int howMuch) {
    // int stillToEnqueue = howMuch;
    // if (howMuch <= 0 || input.size() <= 0) {
    // return 0; // nothing to add
    // }
    //
    // if (queue.size() > 0) {
    // final ByteBuffer last = queue.getLast();
    //
    // if (last.hasRemaining()) {
    // stillToEnqueue -= fillBuffer(input, last, stillToEnqueue);
    // }
    // }
    //
    // while (stillToEnqueue > 0 && input.size() > 0) {
    // ByteBuffer newBuffer = ByteBuffer.allocate(PAYLOAD_BUFFER_SIZE);
    // stillToEnqueue -= fillBuffer(input, newBuffer, stillToEnqueue);
    // queue.add(newBuffer);
    // }
    //
    // return howMuch - stillToEnqueue;
    // }

    // /**
    // * @param input
    // * @param last
    // */
    // private int fillBuffer(InputQueue input, ByteBuffer last, final int
    // howMuch) {
    // final int inputSize = input.size();
    // int written = 0;
    // int howMuchToActuallyTake = inputSize < howMuch ? inputSize : howMuch;
    //
    // if (howMuchToActuallyTake <= last.remaining()) {
    // last.put(input.dequeueBytes(howMuch));
    // written = howMuchToActuallyTake;
    // } else {
    // int remaining = last.remaining();
    // while (last.hasRemaining()) {
    // last.put(input.dequeueBytes(last.remaining()));
    // }
    // written = remaining;
    // }
    // payloadCurrentSize += written;
    // return written;
    // }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.net.cs.server.ProtocolHandler#handleMessage(java
     * .nio.ByteBuffer, org.mentalsmash.tazio.net.cs.server.impl.ChannelFacade)
     */
    @Override
    public final void handleMessage(ByteBuffer message, ChannelFacade source) {
	byte[] messageArray = bufferToArray(message);
	ByteArrayOutputStream backend = null;
	MessageOutputStream messageOutput = null;

	MessageInputStream inputStream = new MessageInputStream(
		new ByteArrayInputStream(messageArray));
	Serializable payload = null;

	try {
	    payload = inputStream.readMessage();

	} catch (ClassCastException e) {
	    String errorMsg = "Received message is not an instance of "
		    + NIOObjectContainer.class.getSimpleName();
	    log.error(errorMsg, e);
	    throw new RuntimeCSSubsystemException(errorMsg, e);
	} catch (ClassNotFoundException e) {
	    log.error("Cannot deserialise object", e);
	    throw new RuntimeCSSubsystemException(e);
	} catch (IOException e) {
	    log.error("Error while retrieving object message. ", e);
	}

	// invokes the higher level handler, passing it the payload and an
	// instance of ObjectChannelWriter
	// to allow applications to write high-level Serializable objects to the
	// channel that originated
	// the request
	backend = new ByteArrayOutputStream();
	messageOutput = new MessageOutputStream(backend, this.getDeflating());

	handleMessage(payload, messageOutput);

	try {
	    messageOutput.flush();
	    // wrapping into a ByteBuffer the contents written by the
	    // MessageOutputStramInstane
	    // and enqueuing them into the output queue of the source
	    if (backend.size() > 0) {
		source.getOutputQueue().enqueue(
			ByteBuffer.wrap(backend.toByteArray()));
	    }
	} catch (IOException e) {
	    log.error("FALTA: Really unexpected excpetion!");
	    throw new RuntimeCSSubsystemException(
		    "This exception should never be thrown", e);
	} finally {
	    try {
		messageOutput.close();
	    } catch (IOException e) {
		// never mind
	    }
	}

    }

    /**
     * This method is invoked by the <i>CS Communication Subsystem</i> each time
     * a Serializable object is received from the channel which the instance of
     * this Handler is associated to. Along with the received message an
     * instance of {@link MessageOutputStream} is provided: use its method
     * {@link MessageOutputStream#writeMessage(Serializable)} to write back an
     * object response to the sender.
     * <p>
     * This method is automatically invoked by the working thread that is
     * associated with an incoming request.
     * </p>
     * 
     * @param msgReceived
     *            the object that was received from the channel associated with
     *            this handler's instance
     * @param channelWriter
     *            an outputStream associated to the channel which sent the
     *            object
     */
    public abstract void handleMessage(Serializable msgReceived,
	    MessageOutputStream channelWriter);

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.net.cs.server.ProtocolHandler#nextMessage(org.
     * mentalsmash.tazio.net.cs.server.impl.ChannelFacade)
     */
    // TODO: Make this method use MessageInputStream for clearenss
    @Override
    public final ByteBuffer nextMessage(ChannelFacade source) {
	InputQueue input = source.getInputQueue();
	ByteBuffer completeMessage;
	// ByteBuffer payloaBuffer;
	int payloadLength;

	try {
	    if (!retrieveHeader(input)) { // tries to retrieve the header of the
		// next message, if it can't returns
		// null
		return null;
	    }
	} catch (IOException e) {
	    throw new RuntimeCSSubsystemException(
		    "FATAL error while retrieving a message hedaer", e);
	}

	payloadLength = this.nextHeader.getPayloadLength();
	if (partialInput == null) {
	    partialInput = ByteBuffer.allocate(payloadLength);
	}

	final int inputSize = input.size();
	if (!input.isEmpty()) {
	    int howMuch = Math.min(payloadLength - partialInput.position(),
		    inputSize);
	    partialInput.put(input.dequeueBytes(howMuch));
	}

	if (partialInput.position() < payloadLength) {
	    return null;
	}

	// When retrieveHeader() has returned true it will have dequeued the
	// header from the input,
	// so there's only the payload to retrieve

	// if (!input.isEmpty()) {
	// enqueue(input, payloadLength); // enqueues at most payloadLength
	// // bytes;
	// }
	//
	// if (payloadCurrentSize < payloadLength) {
	// return null;
	// }

	log.debug("Retrieving msg payload from the channel");

	completeMessage = ByteBuffer.allocate(MessageHeader.HEADER_SIZE
		+ payloadLength); // allocating size for the complete
	// message

	partialInput.flip();

	completeMessage.put(headerBuffer).put(partialInput);

	/*
	 * resetting handler's status
	 */
	this.nextHeader = null;
	headerReceived = false;
	headerBuffer = null;
	partialInput = null;

	return completeMessage;

    }

    // /**
    // * @param completeMessage
    // * @return
    // */
    // private ByteBuffer retrievePayloadFromQueue() {
    // ByteBuffer payloadBuffer = ByteBuffer.allocate(payloadCurrentSize);
    //	
    // ListIterator<ByteBuffer> it = queue.listIterator();
    // while(it.hasNext()) {
    // ByteBuffer buffer = it.next();
    // buffer.flip();
    // payloadBuffer.put(buffer);
    // it.remove();
    // }
    //	
    // payloadBuffer.flip();
    //	
    // return payloadBuffer;
    //	
    // }

    /**
     * Reads from the input source and tries to retrieve a message header. If it
     * manages to find it, dequeues it from the input queue, and changes the
     * status of this protocol handler, putting it in the
     * "Waiting for the payload" state.
     * 
     * @param input
     *            The input queue where to search for the header
     * @return true if the header was previously or just now received. Otherwise
     *         false
     * @throws IOException
     *             If it fails to retrieve a valid header.
     */
    private boolean retrieveHeader(InputQueue input) throws IOException {
	byte[] header;

	if (headerReceived)
	    return true;

	if (input.size() <= MessageHeader.HEADER_SIZE) {
	    return false;
	}

	headerBuffer = input.dequeueBytes(MessageHeader.HEADER_SIZE); // getting
	// the
	// header
	headerReceived = true;

	header = bufferToArray(headerBuffer);

	this.nextHeader = MessageHeader.readHeader(header);

	log.debug("Header for next message completely retrieved.");

	return headerReceived;

    }

    /**
     * Returns the entire content of a byte buffer as an array.
     * <p>
     * If the buffer is backed by an array returns this array: notice that in
     * such a case modifying the array modifies the buffer too. If the buffer is
     * not backed by an array copy the whole buffer content (from the current
     * position to to the current limit) into a newly created array, and then
     * restors the current position
     * </p>
     * 
     * @param b
     *            the buffer to get the array from
     * @return Returns the entire content of a byte buffer as an array.
     */
    private byte[] bufferToArray(ByteBuffer b) {

	if (b.hasArray()) {
	    return b.array();
	} else {
	    b.mark();
	    byte[] array = new byte[b.remaining()];
	    for (int i = 0; b.hasRemaining(); i++) {
		array[i] = b.get();
	    }

	    b.reset(); // reverting the array in read-mode
	    return array;

	}

    }
    
    @Override
    public void setDeflating(boolean compressed){
	this.deflating = compressed;
    }
    
    @Override
    public boolean getDeflating(){
	return this.deflating;
    }

}
